package com.xy.async.mq;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
//import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import com.xy.async.biz.AsyncBizService;
import com.xy.async.constant.AsyncConstant;
import com.xy.async.dto.AsyncExecDto;

import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.Map;

/**
 * 异步执行消费者
 *
 *  
 * @date 2021/01/08
 */
@Slf4j
@Component
public class AsyncConsumer {

    @Autowired
    private AsyncBizService asyncBizService;

    /**
     * 队列名称前缀：默认是应用名称
     */
    @Value("${async.topic:${spring.application.name}}")
    private String asyncTopic;

    /**
     * 消费消息
     *
     * @param asyncExecDto
     * @return
     */
    /*@KafkaListener(topics = "${async.topic:${spring.application.name}}" + AsyncConstant.QUEUE_SUFFIX,groupId = "${spring.application.name}")
    public void onConsume(AsyncExecDto asyncExecDto) {*/
    @RabbitListener(ackMode = "MANUAL",queues = "${async.topic:${spring.application.name}}" + AsyncConstant.QUEUE_SUFFIX,
            concurrency = "20")
    public void  messageConsumer(Message message,AsyncExecDto asyncExecDto, Channel channel,
                                 @Header(AmqpHeaders.DELIVERY_TAG) Long tag ) throws IOException {
        String queueName = asyncTopic + AsyncConstant.QUEUE_SUFFIX;
        try {
            log.info("Rabbit消息开始消费，queueName：'{}'，message：{}", queueName, asyncExecDto);
            // 执行方法
            asyncBizService.invoke(asyncExecDto);
            log.info("Rabbit消息消费成功，queueName：'{}'，message：{}", queueName, asyncExecDto);
            channel.basicAck(tag,false);
        } catch (Exception e) {
            log.error("Rabbit消息消费失败，queueName：'{}'，message：{}", queueName, asyncExecDto, e);
            // ack返回false，requeue-true并重新回到队列
            try {
                channel.basicNack(tag, false, true);
            } catch (IOException ex) {
                log.error("手动ack error",e);
            }
            //channel.basicReject(tag,true);

            //捕获异常后，重新发送到指定队列，自动确认不抛出异常即为ack
            Integer retryCount;
            Map<String, Object> headers = message.getMessageProperties().getHeaders();
            if(!headers.containsKey("retry-count")){
                retryCount=0;
            }else {
                retryCount = (Integer)headers.get("retry-count");
            }
            retryCount = retryCount +1 ;
            log.info("retryCount:{}", retryCount);
            //判断是否满足最大重试次数(重试3次)
            if(retryCount<3) {
                headers.put("retry-count",retryCount);
                //重新发送到MQ中
                AMQP.BasicProperties basicProperties = new AMQP.BasicProperties().builder().contentType("text/plain").headers(headers).build();
                channel.basicPublish(message.getMessageProperties().getReceivedRoutingKey(),
                        message.getMessageProperties().getReceivedRoutingKey(), basicProperties,
                        message.getBody());
            }
           // throw new RuntimeException(e);
        }
    }

}
